package cn.gwm.flink.streaming.strategy.vehiclepredict;

import cn.gwm.flink.streaming.constant.ChargePredictFields;
import com.alibaba.fastjson.JSONObject;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.java.typeutils.MapTypeInfo;
import org.apache.flink.streaming.api.functions.co.KeyedBroadcastProcessFunction;

import java.io.Serializable;
import java.math.BigDecimal;
import java.util.Map;

/**
 * @author : GW00280745
 * @date : 2022/8/15 8:58
 * @description : 间隔点计算 第一个间隔点是首条记录，之后和上一个间隔点进行比较，温度差和电流差符合条件那么就是最新的间隔点
 */
public class MapTempInterval implements Serializable {

    public void process(JSONObject record,
                        MapState<String, String> latestIntervalPointState,
                        KeyedBroadcastProcessFunction.ReadOnlyContext ctx){
        try {
            intervalPoint(record,latestIntervalPointState,ctx);
        }catch (Exception e){
            if (record!=null && record.containsKey(errVin) && record.containsKey(errTid)){
                //出现异常时调用异常处理模块 暂时打印下看看详情 tbd
                System.out.println(record.toJSONString());
            }
            e.printStackTrace();
        }

    }
    private String errVin = "vin";
    private String errTid = "tid";
    private MapStateDescriptor<String, Map<String, Object>> confInfo = new MapStateDescriptor<>("conf-info", BasicTypeInfo.STRING_TYPE_INFO, new MapTypeInfo<>(String.class, Object.class));

    private void intervalPoint(JSONObject line,
                               MapState<String, String> latestIntervalPointState,
                               KeyedBroadcastProcessFunction.ReadOnlyContext ctx) throws Exception{
        //温度间隔点
        String frVal = line.getString(ChargePredictFields.ins().stdFieldChargeInitPoint);
        //首条记录是空值情况下
        if (StringUtils.isBlank(frVal) || StringUtils.isBlank(line.getString(ChargePredictFields.ins().stdFieldTemp))){
            line.put(ChargePredictFields.ins().stdFieldTempIntervalPoint, ChargePredictFields.ins().zeroStr);
            line.put(ChargePredictFields.ins().intervalPoint,ChargePredictFields.ins().blankStr);
            return;
        }
        String chargeGroupName = line.getString(ChargePredictFields.ins().dwsFieldChargeBegin);
        Integer temp = line.getInteger(ChargePredictFields.ins().stdFieldTemp);
        BigDecimal batter = line.getBigDecimal(ChargePredictFields.ins().srcFieldBmsBatterCurrRm);
        Long tidVal = line.getLong(ChargePredictFields.ins().srcFieldTid);
        String s = temp.toString().concat(ChargePredictFields.ins().char3).concat(tidVal.toString()).concat(ChargePredictFields.ins().char3).concat(batter.toString());
        //有首条记录，那么判断状态中是否有数据，如果没有那么首条记录就是第一个间隔点
        if (!latestIntervalPointState.contains(chargeGroupName)){
            latestIntervalPointState.put(chargeGroupName,s);
            line.put(ChargePredictFields.ins().intervalPoint,chargeGroupName);
            line.put(ChargePredictFields.ins().stdFieldTempIntervalPoint,ChargePredictFields.ins().zeroStr);
            return;
        }
        //状态中有数据，那么进行温度和电流比较
        String latestTempBatter = latestIntervalPointState.get(chargeGroupName);
        String[] sp = latestTempBatter.split(ChargePredictFields.ins().char3);
        int latestTemp = Integer.parseInt(sp[0]);
        long latestTid = Long.parseLong(sp[1]);
        BigDecimal latestBatter = new BigDecimal(sp[2]);
        int deltaTemp = Math.abs(temp - latestTemp);
//        double deltaTime = Math.abs(ChargeUtil.single().diffL(tidVal, latestTid))
        double deltaTime = Math.abs(tidVal - latestTid);
        long deltaBatter = Math.abs(batter.longValue() - latestBatter.longValue());
        Map<String, Object> confMap = ctx.getBroadcastState(confInfo).get(null);
        long temRiseTemSection = parseConf(confMap, ChargePredictFields.ins().paramTempAccIntervalTemp, "5");
        long temRiseElectricSection = parseConf(confMap, ChargePredictFields.ins().paramTempAccIntervalBatter, "10");
        //温度=-40是默认值，属于无效值
        boolean b =deltaTemp < temRiseTemSection &&
        deltaBatter < temRiseElectricSection ||
        (deltaTemp == 0) || (temp== -40)
        ;
        if (b){
            line.put(ChargePredictFields.ins().stdFieldTempIntervalPoint,ChargePredictFields.ins().zeroStr);
            line.put(ChargePredictFields.ins().intervalPoint,ChargePredictFields.ins().blankStr);
            return;
        }
        //基于离线统计任务 => 短时间内温度变化剧烈 => 形成规则 = 慢充时间差小于5min的没有意义，快充时间差小于1min的没有意义
        int obcSts = line.getIntValue(ChargePredictFields.ins().srcFieldBmsObcConnect);
        int dcSts = line.getIntValue(ChargePredictFields.ins().srcFieldBmsDcConnect);
        double deltaTimeMinute = (deltaTime / 1000d) / 60d;
        if (deltaTimeMinute == 0){
            return;
        }
        if(obcSts==1 && deltaTimeMinute <= num5){
            return;
        }
        if (dcSts==1 && deltaTimeMinute <= num1f5){
            return;
        }
        Double tempAcc = (double)(temp - latestTemp) / deltaTimeMinute;
        if (Math.abs(tempAcc)> num100){
            return;
        }
//        LogTE.info(String.format("---> 温升速率信息 ###  temp= %s ,latestTemp= %s ,deltaTemp= %s ; batter=%s,latestBatter=%s,deltaBatter=%s;tid= %s ,latestTid= %s ,deltaTime= %s ms/ %s min ,tempAcc = %s ",temp,latestTemp,deltaTemp,batter,latestBatter,deltaBatter,tidVal,latestTid,deltaTime,deltaTimeMinute,tempAcc))
        String accVal = String.format("%.3f", tempAcc).concat(ChargePredictFields.ins().char3).concat(tidVal.toString());
        line.put(ChargePredictFields.ins().tempAccField,tempAcc==null?ChargePredictFields.ins().blankStr:accVal);
        line.put("tempAccVal",tempAcc);
        latestIntervalPointState.put(chargeGroupName,s);
        line.put(ChargePredictFields.ins().stdFieldTempIntervalPoint,ChargePredictFields.ins().oneStr);
        line.put(ChargePredictFields.ins().intervalPoint,chargeGroupName);
    }
    private int num5=5;
    private int num100 = 100;
    private float num1f5 =1.5f;
    private long parseConf(Map<String, Object> confMap,String key,String defaultVal){
        try {
            Object confMapOrDefault = confMap.getOrDefault(key, defaultVal);
            return Long.parseLong(confMapOrDefault.toString());
        }catch (Exception e){
            return Long.parseLong(defaultVal);
        }
    }
}
